package ex.util;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * 
 * @author zy
// * @see  https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/table/tableapi/
 */
public class TableAPI {
	
	public TableEnvironment tEnv ;
	
	public void getEnv() {
		EnvironmentSettings settings = EnvironmentSettings
			    .newInstance()
			    .inStreamingMode()
			    .build();

			this.tEnv = TableEnvironment.create(settings);			
	}

	public Table registerKafka(){
		String createOutDDL="" +
				"CREATE TABLE fKafkaTable (\n" +
				"  `user_id` BIGINT,\n" +
				"  `item_id` BIGINT,\n" +
				"  `behavior` STRING,\n" +
				"  `ts` TIMESTAMP(3) METADATA FROM 'timestamp'\n" +
				") WITH (\n" +
				"  'connector' = 'kafka',\n" +
				"  'topic' = 'user1',\n" +
				"  'properties.bootstrap.servers' = '192.168.1.6:9092',\n" +
				"  'properties.group.id' = 'testGroup',\n" +
				"  'scan.startup.mode' = 'earliest-offset',\n" +
				"  'format' = 'csv'\n" +
				")";
		tEnv.executeSql(createOutDDL);
		Table ft = tEnv.from("fKafkaTable");
		return ft;
	}
	public Table registerMysql(String tablename,String flinkTableName) {
        String createOutDDL = "CREATE TABLE "+flinkTableName+" (" +
                "a STRING, " +
                "b BIGINT, " +
                "c STRING, " +
                "rowtime TIMESTAMP(3)"+
                ") WITH (" +
                "   'connector'  = 'jdbc'," +
                "   'url'        = 'jdbc:mysql://localhost:3306/test?characterEncoding=utf8&serverTimezone=UTC&useSSL=false'," +
                "   'table-name' = '"+ tablename +"'," +
                "   'driver'     = 'com.mysql.cj.jdbc.Driver'," +
                "   'username'   = 'root'," +
                "   'password'   = '12345678'," +
                "   'sink.buffer-flush.max-rows' = '0' "+
        ")";
        tEnv.executeSql(createOutDDL);
        Table ft = tEnv.from(flinkTableName);
        return ft;
	}
   
	
	private void registerConsole() {
	    String createPrintOutDDL = "CREATE TABLE printOutTable (" +
                "a STRING, " +
                "b BIGINT " +
                ") WITH (" +
                " 'connector' = 'print' " +
                ")";
        tEnv.executeSql(createPrintOutDDL);
	}

	public static void main(String[] args) {
		TableAPI demo =new TableAPI();
		demo.getEnv();
		Table orders = demo.registerMysql("orders", "orders");
		 demo.registerConsole();			
		
	      Table counts = orders
	                .groupBy($("a"))
	                .select($("a"), $("b").count().as("cnt"));
   
	        //print  
	        counts.executeInsert("printOutTable");	        
	   	}
}
